package com.xmg.threadpool;

import javafx.concurrent.Worker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class MyThreadPool {
    public static void main(String[] args) {
        /**
         * int corePoolSize,
         *                               int maximumPoolSize,
         *                               long keepAliveTime,
         *                               TimeUnit unit,
         *                               BlockingQueue<Runnable> workQueue,
         *                               ThreadFactory threadFactory,
         *                               RejectedExecutionHandler handler
         */
        new ThreadPoolExecutor(20,100,10, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy() );
        Executors.newFixedThreadPool(5);


        ThreadPool threadPool = new ThreadPool(2,
                1000, TimeUnit.MILLISECONDS, 20, new RejectPolicy<Runnable>() {
            @Override
            public void reject(BlockingQueue<Runnable> queue, Runnable task) {
                //1、死等
                queue.put(task);
                //2、带超时等待
//            queue.offer(task,1500,TimeUnit.MILLISECONDS);
                //3、放弃任务
//            log.info("放弃{}",task);
                //4、抛出异常
//            throw new RuntimeException("任务执行失败 "+task);
                //5、让调用者自己执行
//                task.run();
            }
        });
//        for (int i = 0; i < 500; i++) {
//            final int j = i;
//            ThreadLocal<String> local = new ThreadLocal<>();
//            threadPool.execute(()->{
//                local.set("这是一个模拟地址"+j);
//                try {
//                    Thread.sleep(1);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//                log.info("begin..{}",j);
//            });
//        }

        for (;;){
            ThreadLocal<String> local = new ThreadLocal<>();
            threadPool.execute(()->{
                local.set("这是一个模拟地址"+ UUID.randomUUID().toString());
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}
//拒绝策略
@FunctionalInterface
interface RejectPolicy<T>{
    void reject(BlockingQueue<T> queue,T task);
}

@Slf4j
class ThreadPool{
    //任务队列
    private BlockingQueue<Runnable> taskQueue;
    //线程集合
    private HashSet<Worker> workers = new HashSet<>();
    //核心线程数
    private int coreSize;
    //超时时间
    private long timeout;
    private TimeUnit unit;

    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, long timeout, TimeUnit unit,int queueCapcity,RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.unit = unit;
        this.rejectPolicy =rejectPolicy;
        this.taskQueue = new BlockingQueue<>(queueCapcity);
    }
    //执行任务
    public void execute(Runnable task){
        //当任务数没有超过coreSizes时，交给worker执行
        synchronized (workers){
            if(workers.size()<coreSize){
                Worker worker = new Worker(task);
                log.info("新增worker：{},{}",worker,task);
                workers.add(worker);
                worker.start();
            }else {
                //超过之后，放入任务队列
                //1、死等
                //2、带超时等待
                //3、放弃任务
                //4、抛出异常
                //5、让调用者自己执行
//                taskQueue.put(task);
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }


    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            //1、task不为空，执行任务
            //2、当task执行完毕，从任务队列中获取任务并执行
            while (task!=null ||(task = taskQueue.poll(timeout,unit))!=null){
                try {
                    log.info("正在执行。。{}",task);
                    task.run();
                }catch (Exception e){

                }finally {
                    task = null;
                }
            }

            synchronized (workers){
                log.info("worker被移除：",this);
                workers.remove(this);
            }
        }
    }
}

@Slf4j
class BlockingQueue<T>{
    //1、任务队列
    private Deque<T> queue = new ArrayDeque<>();

    //2、锁
    private ReentrantLock lock = new ReentrantLock();

    //3、生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    //4、消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();
    //5、容量
    private int capcity;

    public BlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    //带超时的阻塞获取
    public T poll(long timeout,TimeUnit unit){
        lock.lock();
        try {
            long nanos = unit.toNanos(timeout);
            while (queue.isEmpty()){
                try {
                    if(nanos<=0){
                        return null;
                    }
                    //添加一个超时时间
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }
    //6、阻塞获取
    public T take(){
        lock.lock();
        try {
            while (queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }
    //7、阻塞添加
    public void put(T element){
        lock.lock();
        try {
            while (queue.size()==capcity){
                try {
                    log.info("等待加入任务队列：{}",element);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            log.info("加入任务队列：{}",element);
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    public boolean offer(T task,long timeout,TimeUnit unit){
        lock.lock();
        try {
            long nanos = unit.toNanos(timeout);
            while (queue.size()==capcity){
                try {
                    log.info("等待加入任务队列：{}",task);
                    if(nanos<=0){
                        return false;
                    }
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            log.info("加入任务队列：{}",task);
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    //8、获取队列大小
    public int size(){
        lock.lock();
        try {
            return queue.size();
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            if(queue.size() == capcity){
                rejectPolicy.reject(this,task);
            }else{
                log.info("加入任务队列：{}",task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}

